package com.om.opensourway;

import Utils.HttpClientUtils;
import com.om.common.CommonObject;
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.broadcast.Broadcast;
import org.codehaus.jackson.JsonNode;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;
import java.util.*;

import static org.elasticsearch.index.query.QueryBuilders.matchAllQuery;

public class PlaygroundLog {
    public static void main(String[] args) throws ClassNotFoundException, IllegalAccessException, InstantiationException {
        String taskCount = args[2];
        String inputPath = args[3];
        String esIndexName = args[4];
        String userIndex = args[5];
        String className = args[7];

        CommonObject commonObject = (CommonObject) Class.forName(className).newInstance();
        commonObject.setEsIndexName(esIndexName);

        SparkConf sparkConf = commonObject.getSparkConf(args);
        JavaSparkContext jsc = new JavaSparkContext(sparkConf);

        Broadcast<CommonObject> broCommonObject = jsc.broadcast(commonObject);
        JavaRDD<String> srcRdd = jsc.textFile(inputPath, Integer.parseInt(taskCount));

        parseUserInfo(srcRdd, userIndex);  // 获取用户信息

        Map<Long, String> providerMap = scrollProvider(userIndex);
        Map<Long, Long> idMap = scrollIds(userIndex);
        Broadcast<Map<Long, Long>> idMapBc = jsc.broadcast(idMap);
        Broadcast<Map<Long, String>> providerMapBc = jsc.broadcast(providerMap);
        parseEventNew(srcRdd, broCommonObject, idMapBc, providerMapBc); // 获取操作事件
        System.out.println("*** PlaygroundLog end ***");
    }

    private static void parseUserInfo(JavaRDD<String> srcRdd, String userIndex) {
        JavaRDD<String> userRDD = srcRdd.filter((Function<String, Boolean>) data -> {
            long userId;
            try {
                String body = CommonObject.objectMapper.readTree(data).get("body").asText();
                if (StringUtils.isBlank(body)) return false;
                JsonNode bodyNode = CommonObject.objectMapper.readTree(body);
                userId = bodyNode.get("UserId").getLongValue();
            } catch (Exception e) {
                return false;
            }
            return userId != 0;
        });

        userRDD.foreachPartition((VoidFunction<Iterator<String>>) it -> {
            RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
            BulkRequest request = new BulkRequest();

            while (it.hasNext()) {
                String data = it.next();
                String body = CommonObject.objectMapper.readTree(data).get("body").asText();
                JsonNode bodyNode = CommonObject.objectMapper.readTree(body);
                Map map = CommonObject.objectMapper.readValue(body, Map.class);
                long id = bodyNode.get("UserId").getLongValue();
                map.put("authing_id", Long.parseLong(bodyNode.get("SubUid").getTextValue()));
                request.add(new IndexRequest(userIndex, "_doc", String.valueOf(id)).source(map));
            }
            if (request.requests().size() != 0) restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            restHighLevelClient.close();
        });
    }

    private static Map<Long, Long> scrollIds(String indexName) {
        RestHighLevelClient client = HttpClientUtils.restClient();

        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(matchAllQuery());
        searchSourceBuilder.size(10000); //设定每次返回多少条数据
        searchSourceBuilder.fetchSource(new String[]{"UserId", "authing_id"}, null); //设置返回字段和排除字段

        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .scroll(scroll)
                .source(searchSourceBuilder);

        HashMap<Long, Long> idMap = new HashMap<>();
        String scrollId = null;
        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            for (SearchHit searchHit : searchHits) {
                Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                idMap.put(Long.valueOf((Integer) sourceAsMap.get("UserId")), Long.valueOf((Integer) sourceAsMap.get("authing_id")));
            }
            while (searchHits != null && searchHits.length > 0) {
                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                scrollId = searchResponse.getScrollId();
                searchHits = searchResponse.getHits().getHits();
                if (searchHits != null && searchHits.length > 0) {
                    for (SearchHit searchHit : searchHits) {
                        Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                        idMap.put(Long.valueOf((Integer) sourceAsMap.get("UserId")), Long.valueOf((Integer) sourceAsMap.get("authing_id")));
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //clean scroll
            if (scrollId != null) {
                try {
                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                    clearScrollRequest.addScrollId(scrollId);
                    ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                    boolean succeeded = clearScrollResponse.isSucceeded();
                    client.close();
                    System.out.printf("*** clear scroll: %s ***%n", succeeded);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return idMap;
    }

    private static Map<Long, String> scrollProvider(String indexName) {
        RestHighLevelClient client = HttpClientUtils.restClient();

        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(matchAllQuery());
        searchSourceBuilder.size(10000); //设定每次返回多少条数据
        searchSourceBuilder.fetchSource(new String[]{"UserId", "Identity"}, null); //设置返回字段和排除字段

        SearchRequest searchRequest = new SearchRequest()
                .indices(indexName)
                .scroll(scroll)
                .source(searchSourceBuilder);

        HashMap<Long, String> idMap = new HashMap<>();
        String scrollId = null;
        try {
            SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            SearchHit[] searchHits = searchResponse.getHits().getHits();
            for (SearchHit searchHit : searchHits) {
                Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                List<Map> identity = (List<Map>) sourceAsMap.get("Identity");
                if (identity == null) continue;
                for (Map map : identity) {
                    String provider = (String) map.getOrDefault("Provider", "");
                    if (StringUtils.isNotBlank(provider)) {
                        idMap.put(Long.valueOf((Integer) sourceAsMap.get("UserId")), provider);
                    }
                }
            }
            while (searchHits != null && searchHits.length > 0) {
                SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                scrollRequest.scroll(scroll);
                searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                scrollId = searchResponse.getScrollId();
                searchHits = searchResponse.getHits().getHits();
                if (searchHits != null && searchHits.length > 0) {
                    for (SearchHit searchHit : searchHits) {
                        Map<String, Object> sourceAsMap = searchHit.getSourceAsMap();
                        List<Map> identity = (List<Map>) sourceAsMap.get("Identity");
                        for (Map map : identity) {
                            String provider = (String) map.getOrDefault("Provider", "");
                            if (StringUtils.isNotBlank(provider)) {
                                idMap.put(Long.valueOf((Integer) sourceAsMap.get("UserId")), provider);
                            }
                        }
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //clean scroll
            if (scrollId != null) {
                try {
                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                    clearScrollRequest.addScrollId(scrollId);
                    ClearScrollResponse clearScrollResponse = client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                    boolean succeeded = clearScrollResponse.isSucceeded();
                    client.close();
                    System.out.printf("*** clear scroll: %s ***%n", succeeded);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return idMap;
    }

    private static void parseEvent(JavaRDD<String> srcRdd, Broadcast<CommonObject> broCommonObject, Broadcast<Map<Long, Long>> idMapBc) {
        srcRdd.foreachPartition((VoidFunction<Iterator<String>>) it -> {
            RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
            BulkRequest request = new BulkRequest();

            Map<Long, Long> idMap = idMapBc.value();
            while (it.hasNext()) {
                String data = it.next();
                Map map = CommonObject.objectMapper.readValue(data, Map.class);
                map.remove("body");
                map.remove("time");

                long userId = Long.parseLong(map.get("userId").toString());
                long authingId = idMap.getOrDefault(userId, 0L);
                map.put("authing_id", authingId);

                String operationTime = map.get("operationTime").toString();
                String createdAt = operationTime.replace(" ", "T") + "+08:00";
                map.put("created_at", createdAt);

                String chapterDur = map.get("chapterDur").toString();
                long chapterDurValue = chapterDur.equals("") ? 0L : Long.parseLong(chapterDur.replace(" min", ""));
                map.put("chapterDur", chapterDurValue);

                String courseDur = map.get("courseDur").toString();
                long courseDurValue = chapterDur.equals("") ? 0L : Long.parseLong(courseDur);
                map.put("courseDur", courseDurValue);

                request.add(new IndexRequest(broCommonObject.value().getEsIndexName(), "_doc", DigestUtils.md5Hex(data)).source(map));
            }
            if (request.requests().size() != 0) restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            restHighLevelClient.close();
        });
        System.out.println("*** Event end ***");
    }

    private static void parseEventNew(JavaRDD<String> srcRdd, Broadcast<CommonObject> broCommonObject, Broadcast<Map<Long, Long>> idMapBc, Broadcast<Map<Long, String>> providerMapBc) {
        srcRdd.foreachPartition((VoidFunction<Iterator<String>>) it -> {
            RestHighLevelClient restHighLevelClient = HttpClientUtils.restClient();
            BulkRequest request = new BulkRequest();

            Map<Long, Long> idMap = idMapBc.value();
            Map<Long, String> providerMap = providerMapBc.value();
            while (it.hasNext()) {
                String data = it.next();

                Map map = CommonObject.objectMapper.readValue(data, Map.class);

                HashMap<String, Object> resMap = new HashMap<>();
                resMap.put("appId", map.get("appId"));
                resMap.put("eventType", map.get("eventType"));

                String operationTime = map.get("operationTime").toString();
                String createdAt = operationTime.replace(" ", "T") + "+08:00";
                resMap.put("operationTime", createdAt);

                resMap.put("state", map.get("state"));
                resMap.put("stateMessage", map.get("stateMessage"));

                long userId = Long.parseLong(map.get("userId").toString());
                resMap.put("userProvider", providerMap.getOrDefault(userId, ""));

                long authingId = idMap.getOrDefault(userId, 0L);
                resMap.put("userId", String.valueOf(authingId));

                HashMap<Object, Object> bodyMap = new HashMap<>();
                bodyMap.put("chapterId", map.get("chapterId"));
                bodyMap.put("chapterName", map.get("chapterName"));
                bodyMap.put("chapterStatus", map.get("chapterStatus"));
                String chapterDur = map.get("chapterDur").toString();
                long chapterDurValue = chapterDur.equals("") ? 0L : Long.parseLong(chapterDur.replace(" min", ""));
                bodyMap.put("chapterDur", chapterDurValue);

                bodyMap.put("courseId", map.get("courseId"));
                bodyMap.put("courseName", map.get("courseName"));
                bodyMap.put("courseStatus", map.get("courseStatus"));
                String courseDur = map.get("courseDur").toString();
                long courseDurValue = chapterDur.equals("") ? 0L : Long.parseLong(courseDur);
                bodyMap.put("courseDur", courseDurValue);

                resMap.put("body", bodyMap);

                request.add(new IndexRequest(broCommonObject.value().getEsIndexName(), "_doc", DigestUtils.md5Hex(data)).source(resMap));
            }
            if (request.requests().size() != 0) restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
            restHighLevelClient.close();
        });
        System.out.println("*** Event end ***");
    }
}
